package com.shihb.akka.sparkmasrterwork.master

import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import com.shihb.akka.sparkmasrterwork.common.{HeartBeat, RegisterWorkerInfo, RemoveTimeOutWorker, SendHeartBeat, StartTimeOutWorker, WorkerInfo}
import com.typesafe.config.{Config, ConfigFactory}

import scala.collection.mutable

/**
 * Description:
 *
 * @author shihb
 * @date 2020/10/3 10:56
 */
class SparkMaster extends Actor {
  val workers: mutable.Map[String, WorkerInfo] = mutable.Map[String, WorkerInfo]()

  override def receive: Receive = {
    case "start" => {//开始
      println("master working...")
      self ! StartTimeOutWorker
    }
    case RegisterWorkerInfo(id, cpu, ram) => {//收到worker注册信息
      if (!workers.contains(id)) {
        workers += (id -> WorkerInfo(id, cpu, ram))
        println("workers",workers)
        //回复消息
        sender() ! RegisterWorkerInfo
      }

    }
    case HeartBeat(id)=>{//收到worker心跳信息
      workers(id).lastHeartBeat=System.currentTimeMillis();
      println(s"worker:$id 更新了心跳时间${workers(id).lastHeartBeat}")
    }
    case StartTimeOutWorker=>{//触发开始心跳检查的协议
      println("开始定时检测心跳的程序")
      import scala.concurrent.duration._
      import context.dispatcher
      //说明 0 millis不延时,3 second每隔9秒,self 发给给自己，发送的消息
      context.system.scheduler.schedule(0 millis,9 second,self,RemoveTimeOutWorker)
    }
    case RemoveTimeOutWorker=>{//收到检测心跳
      // 首先获取所有workInfo
      val workInfos: Iterable[WorkerInfo] = workers.values
      val now: Long = System.currentTimeMillis()
      // 把所有超时的删除即可
      workInfos.filter(workerInfo => (now - workerInfo.lastHeartBeat) > 6000)
        .foreach(worker=>workers.remove(worker.id))
      println(s"当前还有${workers.size}个worker存活")
    }
  }
}

object SparkMaster {
  def main(args: Array[String]): Unit = {
    if(args.length!=3){
      println("请输入参数host port sparkMasterActor名字")
      sys.exit()
    }
    val host=args(0)
    val port=args(1)
    val name=args(2)


    //创建ActorSystem
    val config: Config = ConfigFactory.parseString(
      s"""
         |akka.actor.provider="akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname=$host
         |akka.remote.netty.tcp.port=${port.toInt}
         |""".stripMargin)
    val masterActorSystem: ActorSystem = ActorSystem("SparkMaster", config)
    // 创建Actor的引用
    val actorRef: ActorRef = masterActorSystem.actorOf(Props[SparkMaster], name)
    // 启动
    actorRef ! "start"
  }

}
